feat: add Jungian Intelligence Layer service v3.2.0#5
Conversation
Add JungianService and AsyncJungianService with 5 methods each: search_with_context, get_individuation_score, get_individuation_history, get_synchronicity_patterns, get_dream_history. Includes 7 new Pydantic models for archetypes, individuation, synchronicity, and dream data.
Reviewer's GuideAdds a synchronous and asynchronous Jungian Intelligence Layer service ( Sequence diagram for JungianService search_with_context and individuation callssequenceDiagram
actor Developer
participant PluggedInClient
participant JungianService
participant HTTP_API as HTTP_API_Server
Developer->>PluggedInClient: create PluggedInClient(api_key)
PluggedInClient->>JungianService: JungianService(self)
activate JungianService
JungianService-->>PluggedInClient: instance with client.jungian
deactivate JungianService
Developer->>JungianService: search_with_context(query, tool_name, outcome)
activate JungianService
JungianService->>PluggedInClient: request("POST", "/api/memory/archetype/inject", json=payload)
activate PluggedInClient
activate HTTP_API
PluggedInClient->>HTTP_API: POST /api/memory/archetype/inject
HTTP_API-->>PluggedInClient: 200 OK, JSON patterns
deactivate HTTP_API
PluggedInClient-->>JungianService: Response
deactivate PluggedInClient
JungianService-->>Developer: ArchetypeSearchResponse
deactivate JungianService
Developer->>JungianService: get_individuation_score()
activate JungianService
JungianService->>PluggedInClient: request("GET", "/api/memory/individuation")
activate PluggedInClient
activate HTTP_API
PluggedInClient->>HTTP_API: GET /api/memory/individuation
HTTP_API-->>PluggedInClient: 200 OK, JSON individuation
deactivate HTTP_API
PluggedInClient-->>JungianService: Response
deactivate PluggedInClient
JungianService-->>Developer: IndividuationResponse
deactivate JungianService
Class diagram for Jungian Intelligence Layer services and modelsclassDiagram
class PluggedInClient {
+ClipboardService clipboard
+DocumentService documents
+JungianService jungian
+RagService rag
+UploadService uploads
+AgentService agents
+request(method, path, params, json)
}
class AsyncPluggedInClient {
+AsyncClipboardService clipboard
+AsyncDocumentService documents
+AsyncJungianService jungian
+AsyncRagService rag
+AsyncUploadService uploads
+AsyncAgentService agents
+request(method, path, params, json)
}
class JungianService {
-PluggedInClient client
+JungianService(client)
+search_with_context(query, tool_name, outcome) ArchetypeSearchResponse
+get_individuation_score() IndividuationResponse
+get_individuation_history(days) List~IndividuationHistoryEntry~
+get_synchronicity_patterns() List~SynchronicityPattern~
+get_dream_history() List~DreamConsolidation~
}
class AsyncJungianService {
-AsyncPluggedInClient client
+AsyncJungianService(client)
+search_with_context(query, tool_name, outcome) ArchetypeSearchResponse
+get_individuation_score() IndividuationResponse
+get_individuation_history(days) List~IndividuationHistoryEntry~
+get_synchronicity_patterns() List~SynchronicityPattern~
+get_dream_history() List~DreamConsolidation~
}
class IndividuationComponents {
+int memory_depth
+int learning_velocity
+int collective_contribution
+int self_awareness
}
class IndividuationResponse {
+int total
+str level
+str weekly_trend
+str tip
+IndividuationComponents components
}
class IndividuationHistoryEntry {
+str date
+int total
+str level
+IndividuationComponents components
}
class ArchetypedPattern {
+str uuid
+str archetype
+str archetype_label
+float archetype_weight
+str pattern_type
+str description
+str pattern
+float confidence
+float similarity
}
class ArchetypeSearchResponse {
+List~ArchetypedPattern~ patterns
}
class SynchronicityPattern {
+str uuid
+str pattern_type
+str description
+float confidence
+int unique_profiles
}
class DreamConsolidation {
+str uuid
+int source_count
+int token_savings
+float cluster_similarity
+str created_at
}
PluggedInClient --> JungianService : owns
AsyncPluggedInClient --> AsyncJungianService : owns
IndividuationResponse --> IndividuationComponents : has
IndividuationHistoryEntry --> IndividuationComponents : has
ArchetypeSearchResponse --> ArchetypedPattern : contains
JungianService --> IndividuationResponse : returns
JungianService --> IndividuationHistoryEntry : returns
JungianService --> ArchetypeSearchResponse : returns
JungianService --> SynchronicityPattern : returns
JungianService --> DreamConsolidation : returns
AsyncJungianService --> IndividuationResponse : returns
AsyncJungianService --> IndividuationHistoryEntry : returns
AsyncJungianService --> ArchetypeSearchResponse : returns
AsyncJungianService --> SynchronicityPattern : returns
AsyncJungianService --> DreamConsolidation : returns
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
Summary of ChangesHello, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request integrates a new Jungian Intelligence Layer into the SDK, providing advanced capabilities for psychological development analysis, archetype-aware pattern recognition, and memory consolidation. This enhancement allows clients to interact with a sophisticated AI layer designed to understand and process information with a deeper, more nuanced context, significantly expanding the SDK's analytical and interpretive functionalities. Highlights
Changelog
Activity
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here. You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension. Footnotes
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 1 issue, and left some high level feedback:
- The
_parse_*helper functions currently return empty lists when the response shape is unexpected; consider raising a domain-specific error instead so API shape changes are surfaced clearly instead of failing silently. - The helper parsers (
_parse_individuation_history,_parse_synchronicity_patterns,_parse_dream_history) are all typed asAnyfor the input data; tightening these toUnion[Dict[str, Any], List[Dict[str, Any]]](or similar) would make mypy and IDE support more accurate and catch misuse earlier.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The `_parse_*` helper functions currently return empty lists when the response shape is unexpected; consider raising a domain-specific error instead so API shape changes are surfaced clearly instead of failing silently.
- The helper parsers (`_parse_individuation_history`, `_parse_synchronicity_patterns`, `_parse_dream_history`) are all typed as `Any` for the input data; tightening these to `Union[Dict[str, Any], List[Dict[str, Any]]]` (or similar) would make mypy and IDE support more accurate and catch misuse earlier.
## Individual Comments
### Comment 1
<location path="src/pluggedinkit/services/jungian.py" line_range="41" />
<code_context>
+ return IndividuationResponse(**data)
+
+
+def _parse_individuation_history(data: Any) -> List[IndividuationHistoryEntry]:
+ """Parse the individuation history response into typed list."""
+ if isinstance(data, list):
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting shared list-extraction and request-building helpers to reduce repeated parsing and wiring logic across the Jungian service functions.
You can trim complexity a bit in two focused areas without changing any behavior.
### 1. Centralize the “flexible shape” parsing
The three helpers `_parse_individuation_history`, `_parse_synchronicity_patterns`, and `_parse_dream_history` all repeat the same pattern: accept `Any`, handle `list` or `dict` with a couple of possible keys, and silently fall back to `[]`.
You can keep the same accepted shapes but centralize the logic so each parser is simpler and less branched:
```python
from typing import Callable, Type, TypeVar
T = TypeVar("T")
def _extract_list_payload(
data: Any,
*,
default_key: str,
alt_key: Optional[str] = None,
) -> List[Dict[str, Any]]:
if isinstance(data, list):
return data
if isinstance(data, dict):
if default_key in data and isinstance(data[default_key], list):
return data[default_key]
if alt_key and alt_key in data and isinstance(data[alt_key], list):
return data[alt_key]
# Preserve current behavior: unexpected shape -> empty list
return []
```
Then the individual parsers become much narrower and easier to reason about:
```python
def _parse_individuation_history(data: Any) -> List[IndividuationHistoryEntry]:
entries = _extract_list_payload(data, default_key="history", alt_key="entries")
return [IndividuationHistoryEntry(**entry) for entry in entries]
def _parse_synchronicity_patterns(data: Any) -> List[SynchronicityPattern]:
patterns = _extract_list_payload(data, default_key="patterns")
return [SynchronicityPattern(**p) for p in patterns]
def _parse_dream_history(data: Any) -> List[DreamConsolidation]:
entries = _extract_list_payload(data, default_key="history", alt_key="consolidations")
return [DreamConsolidation(**d) for d in entries]
```
If you want to make unexpected shapes more visible later, you can swap the `return []` in `_extract_list_payload` for a `PluggedInError` without changing all call sites.
---
### 2. Share sync/async request + parse wiring
The sync and async services mostly differ in:
- the client type, and
- whether `request` is awaited.
You can keep the separate public classes/signatures but factor out the “request + parse” pattern into a small helper to cut repetition. For example, for the individuation endpoints:
```python
# shared helper
def _build_individuation_history_params(days: int) -> Dict[str, str]:
return {"history": "true", "days": str(days)}
```
Usage in sync:
```python
def get_individuation_history(self, days: int = 30) -> List[IndividuationHistoryEntry]:
response = self.client.request(
"GET",
"/api/memory/individuation",
params=_build_individuation_history_params(days),
)
return _parse_individuation_history(response.json())
```
Usage in async:
```python
async def get_individuation_history(self, days: int = 30) -> List[IndividuationHistoryEntry]:
response = await self.client.request(
"GET",
"/api/memory/individuation",
params=_build_individuation_history_params(days),
)
return _parse_individuation_history(response.json())
```
You can apply the same approach for:
- `_build_search_with_context_payload(query, tool_name, outcome)`
- constant paths like `"/api/memory/individuation"` / `"/api/memory/sync/patterns"` / `"/api/memory/dream/history"`
This keeps functionality identical while reducing duplicated literals and argument construction logic, and makes future changes (e.g., adding a param) less error-prone across sync/async variants.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| return IndividuationResponse(**data) | ||
|
|
||
|
|
||
| def _parse_individuation_history(data: Any) -> List[IndividuationHistoryEntry]: |
There was a problem hiding this comment.
issue (complexity): Consider extracting shared list-extraction and request-building helpers to reduce repeated parsing and wiring logic across the Jungian service functions.
You can trim complexity a bit in two focused areas without changing any behavior.
1. Centralize the “flexible shape” parsing
The three helpers _parse_individuation_history, _parse_synchronicity_patterns, and _parse_dream_history all repeat the same pattern: accept Any, handle list or dict with a couple of possible keys, and silently fall back to [].
You can keep the same accepted shapes but centralize the logic so each parser is simpler and less branched:
from typing import Callable, Type, TypeVar
T = TypeVar("T")
def _extract_list_payload(
data: Any,
*,
default_key: str,
alt_key: Optional[str] = None,
) -> List[Dict[str, Any]]:
if isinstance(data, list):
return data
if isinstance(data, dict):
if default_key in data and isinstance(data[default_key], list):
return data[default_key]
if alt_key and alt_key in data and isinstance(data[alt_key], list):
return data[alt_key]
# Preserve current behavior: unexpected shape -> empty list
return []Then the individual parsers become much narrower and easier to reason about:
def _parse_individuation_history(data: Any) -> List[IndividuationHistoryEntry]:
entries = _extract_list_payload(data, default_key="history", alt_key="entries")
return [IndividuationHistoryEntry(**entry) for entry in entries]
def _parse_synchronicity_patterns(data: Any) -> List[SynchronicityPattern]:
patterns = _extract_list_payload(data, default_key="patterns")
return [SynchronicityPattern(**p) for p in patterns]
def _parse_dream_history(data: Any) -> List[DreamConsolidation]:
entries = _extract_list_payload(data, default_key="history", alt_key="consolidations")
return [DreamConsolidation(**d) for d in entries]If you want to make unexpected shapes more visible later, you can swap the return [] in _extract_list_payload for a PluggedInError without changing all call sites.
2. Share sync/async request + parse wiring
The sync and async services mostly differ in:
- the client type, and
- whether
requestis awaited.
You can keep the separate public classes/signatures but factor out the “request + parse” pattern into a small helper to cut repetition. For example, for the individuation endpoints:
# shared helper
def _build_individuation_history_params(days: int) -> Dict[str, str]:
return {"history": "true", "days": str(days)}Usage in sync:
def get_individuation_history(self, days: int = 30) -> List[IndividuationHistoryEntry]:
response = self.client.request(
"GET",
"/api/memory/individuation",
params=_build_individuation_history_params(days),
)
return _parse_individuation_history(response.json())Usage in async:
async def get_individuation_history(self, days: int = 30) -> List[IndividuationHistoryEntry]:
response = await self.client.request(
"GET",
"/api/memory/individuation",
params=_build_individuation_history_params(days),
)
return _parse_individuation_history(response.json())You can apply the same approach for:
_build_search_with_context_payload(query, tool_name, outcome)- constant paths like
"/api/memory/individuation"/"/api/memory/sync/patterns"/"/api/memory/dream/history"
This keeps functionality identical while reducing duplicated literals and argument construction logic, and makes future changes (e.g., adding a param) less error-prone across sync/async variants.
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a new JungianService for both synchronous and asynchronous clients. The implementation is well-structured, adding new service methods, corresponding Pydantic types, and integrating them into the client and package exports.
My review includes a few suggestions for improvement:
- In
types.py, I've recommended updating the new Pydantic models to usemodel_configinstead of the deprecatedclass Configfor consistency with Pydantic v2. - In
services/jungian.py, I've identified a robustness issue in the JSON parsing helper functions. They could fail if the API returns an unexpected data type. I've provided suggestions to make them more resilient. - I've also added a comment regarding the code duplication between the synchronous and asynchronous service classes, suggesting potential strategies to improve maintainability.
| if isinstance(data, list): | ||
| return [IndividuationHistoryEntry(**entry) for entry in data] | ||
| # If wrapped in an object, try the 'history' key | ||
| if isinstance(data, dict): | ||
| entries = data.get("history", data.get("entries", [])) | ||
| return [IndividuationHistoryEntry(**entry) for entry in entries] | ||
| return [] |
There was a problem hiding this comment.
The current implementation is not fully robust. If the API returns an unexpected data type (e.g., None or a string instead of a list for history or entries), the function will raise a TypeError. It's safer to validate that the extracted data is a list before iterating over it and that each item is a dictionary before unpacking.
items_raw = []
if isinstance(data, list):
items_raw = data
elif isinstance(data, dict):
items_raw = data.get("history")
if not isinstance(items_raw, list):
items_raw = data.get("entries")
if not isinstance(items_raw, list):
return []
return [IndividuationHistoryEntry(**entry) for entry in items_raw if isinstance(entry, dict)]| if isinstance(data, list): | ||
| return [SynchronicityPattern(**p) for p in data] | ||
| if isinstance(data, dict): | ||
| patterns = data.get("patterns", []) | ||
| return [SynchronicityPattern(**p) for p in patterns] | ||
| return [] |
There was a problem hiding this comment.
Similar to the other parsing functions in this file, this implementation is not robust against unexpected API response formats. If data.get('patterns') returns a non-list iterable or None, this function will raise a TypeError. The implementation should be updated to handle these cases gracefully.
| if isinstance(data, list): | |
| return [SynchronicityPattern(**p) for p in data] | |
| if isinstance(data, dict): | |
| patterns = data.get("patterns", []) | |
| return [SynchronicityPattern(**p) for p in patterns] | |
| return [] | |
| items_raw = [] | |
| if isinstance(data, list): | |
| items_raw = data | |
| elif isinstance(data, dict): | |
| items_raw = data.get("patterns") | |
| if not isinstance(items_raw, list): | |
| return [] | |
| return [SynchronicityPattern(**p) for p in items_raw if isinstance(p, dict)] |
| if isinstance(data, list): | ||
| return [DreamConsolidation(**d) for d in data] | ||
| if isinstance(data, dict): | ||
| entries = data.get("history", data.get("consolidations", [])) | ||
| return [DreamConsolidation(**d) for d in entries] | ||
| return [] |
There was a problem hiding this comment.
This function has the same robustness issue as the other parsers. It can fail with a TypeError if the API response for history or consolidations is not a list. It's important to validate the data type before iteration to prevent runtime errors.
| if isinstance(data, list): | |
| return [DreamConsolidation(**d) for d in data] | |
| if isinstance(data, dict): | |
| entries = data.get("history", data.get("consolidations", [])) | |
| return [DreamConsolidation(**d) for d in entries] | |
| return [] | |
| items_raw = [] | |
| if isinstance(data, list): | |
| items_raw = data | |
| elif isinstance(data, dict): | |
| items_raw = data.get("history") | |
| if not isinstance(items_raw, list): | |
| items_raw = data.get("consolidations") | |
| if not isinstance(items_raw, list): | |
| return [] | |
| return [DreamConsolidation(**d) for d in items_raw if isinstance(d, dict)] |
| # ----------------------------------------------------------------------------- | ||
|
|
||
|
|
||
| class AsyncJungianService: |
There was a problem hiding this comment.
The AsyncJungianService class is almost an exact copy of JungianService, with the only differences being async and await. This creates significant code duplication across all five methods, making future maintenance harder (e.g., adding a parameter requires changes in two places). While this pattern might be consistent with other services in the SDK, it's worth considering strategies to reduce this duplication for better maintainability. For example, a tool like unasync could generate the sync code from the async code, or parameter-building logic could be extracted into shared helper functions.
| class Config: | ||
| populate_by_name = True |
There was a problem hiding this comment.
The project appears to be using Pydantic v2 (based on the use of @model_validator elsewhere). For new models, it's better to use model_config instead of the deprecated class Config for configuration. This improves consistency and future-proofs the code. This comment applies to all new Pydantic models in this file (IndividuationResponse, IndividuationHistoryEntry, ArchetypedPattern, SynchronicityPattern, DreamConsolidation).
model_config = {"populate_by_name": True}
Code Review -- feat: add Jungian Intelligence Layer serviceOverall the PR follows the existing patterns well -- clean sync/async symmetry, proper use of TYPE_CHECKING, and thorough docstrings. Here are the issues found, from most to least critical. Bugs1. Unused import + missing API error checking (jungian.py:10) PluggedInError is imported but never used in any method. Compare with clipboard.py which explicitly checks the success field before parsing: If the API returns an error body, callers get a confusing Pydantic ValidationError instead of the documented PluggedInError. Either add a success check or remove the unused import -- the current state gives a false impression that errors are handled. 2. Possible camelCase mismatch in search_with_context payload (jungian.py:111) The optional field is sent as snake_case: payload[tool_name] = tool_name. Every other JSON payload in this SDK uses camelCase (e.g. createdByTool, ttlSeconds). If the API expects toolName, this field is silently ignored, causing subtle context-injection failures. Please verify the expected API field name. Type / Model Issues3. DreamConsolidation.created_at and IndividuationHistoryEntry.date should be datetime (types.py:418, 468) The rest of the models use datetime for timestamps (Document.created_at, ClipboardEntry.created_at). Using str is inconsistent and loses Pydantic's automatic ISO-8601 parsing. 4. No field constraints on score components (types.py:395-403) IndividuationComponents fields have no bounds. The existing codebase shows the pattern (e.g. temperature: Optional[float] = Field(None, ge=0, le=2)). If scores have a fixed range (e.g. 0-100), add ge/le constraints. Code Quality**5. Prefer model_validate() over Model(data) for Pydantic v2 (jungian.py:32, 38) Model(**data) works but Model.model_validate(data) is the idiomatic Pydantic v2 API. It handles nested coercion more explicitly and reliably. 6. _parse_archetype_search manual iteration is inconsistent (jungian.py:29-33) The function manually iterates and constructs ArchetypedPattern instances, while _parse_individuation simply delegates to the model constructor. Pydantic already handles nested list coercion, so ArchetypeSearchResponse.model_validate(data) is sufficient and consistent. 7. No input validation on the days parameter (jungian.py:136, 252) There is no guard against days <= 0. A simple check matches the defensive style used elsewhere in the SDK. Other8. Version not bumped in pyproject.toml The PR title says v3.2.0 but pyproject.toml still reads version = 1.0.1. Adding a new public service warrants at least a minor version bump (1.1.0 under semver). 9. No tests The test plan checkboxes are all unchecked and no tests/ directory exists, despite the project having pytest, pytest-asyncio, and pytest-cov as dev dependencies. Unit tests with mocked HTTP responses for each method (sync and async) would be a good addition before merging. What is done well
|
1 participant
Summary
JungianServiceandAsyncJungianServiceexposed asclient.jungian.*search_with_context(),get_individuation_score(),get_individuation_history(),get_synchronicity_patterns(),get_dream_history()Test plan
Summary by Sourcery
Add synchronous and asynchronous Jungian Intelligence Layer services to the client, exposing archetype-aware memory search, individuation scoring, synchronicity patterns, and dream history APIs.
New Features:
Enhancements: